package chapter_6.transformer;


import model.StockPerformance;
import model.StockTransaction;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.PunctuationType;
import org.apache.kafka.streams.processor.Punctuator;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.KeyValueStore;

import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;


/**
 * The {@code Transformer} interface is for stateful mapping of an input record to zero, one, or multiple new output
 * records (both key and value type can be altered arbitrarily).
 * This is a stateful record-by-record operation, i.e, {@link # transform(Object, Object)} is invoked individually for
 * each record of a stream and can access and modify a state that is available beyond a single call of
 * {@link # transform(Object, Object)} (cf. {@link KeyValueMapper} for stateless record transformation).
 * Additionally, this {@code Transformer} can {@link ProcessorContext#schedule(Duration, PunctuationType, Punctuator) schedule}
 * a method to be {@link Punctuator#punctuate(long) called periodically} with the provided context.
 * <p>
 * Use {@link TransformerSupplier} to provide new instances of {@code Transformer} to Kafka Stream's runtime.
 * <p>
 * If only a record's value should be modified {@link ValueTransformer} can be used.
 *
 * @param # String key type
 * @param # StockTransaction value type
 * @param # KeyValue<String, List<KeyValue<String, StockPerformance>>>
 *     {@link KeyValue} return type (both key and value type can be set
 *            arbitrarily)
 * @see TransformerSupplier
 * @see KStream#transform(TransformerSupplier, String...)
 * @see ValueTransformer
 * @see KStream#map(KeyValueMapper)
 * @see KStream#flatMap(KeyValueMapper)
 */
public class StockPerformanceMultipleValuesTransformer
        implements Transformer<String, StockTransaction,
            KeyValue<String, List<KeyValue<String, StockPerformance>>>> {

    private String stateStoreName ;
    private double differentialThreshold = 0.02;
    private ProcessorContext processorContext;
    private KeyValueStore<String, StockPerformance> keyValueStore;


    public StockPerformanceMultipleValuesTransformer(String stateStoreName, double differentialThreshold) {
        this.stateStoreName = stateStoreName;
        this.differentialThreshold = differentialThreshold;
    }


    @SuppressWarnings("unchecked")
    @Override
    public void init(ProcessorContext processorContext) {
        this.processorContext = processorContext;
        keyValueStore = (KeyValueStore) this.processorContext.getStateStore(stateStoreName);
        this.processorContext.schedule(15000, PunctuationType.STREAM_TIME, this::punctuate);
    }

    @Override
    public KeyValue<String, List<KeyValue<String, StockPerformance>>> transform(
            String symbol, StockTransaction transaction) {
        if (symbol != null) {
            StockPerformance stockPerformance = keyValueStore.get(symbol);

            if (stockPerformance == null) {
                stockPerformance = new StockPerformance();
            }

            stockPerformance.updatePriceStats(transaction.getSharePrice());
            stockPerformance.updateVolumeStats(transaction.getShares());
            stockPerformance.setLastUpdateSent(Instant.now());

            keyValueStore.put(symbol, stockPerformance);
        }
        return null;
    }

    public KeyValue<String, List<KeyValue<String, StockPerformance>>> punctuate(long timestamp) {
        List<KeyValue<String, StockPerformance>> stockPerformanceList = new ArrayList<>();
        KeyValueIterator<String, StockPerformance> performanceIterator = keyValueStore.all();
        while (performanceIterator.hasNext()) {
            KeyValue<String, StockPerformance> keyValue = performanceIterator.next();
            StockPerformance stockPerformance = keyValue.value;

            if (stockPerformance != null) {
                if (stockPerformance.priceDifferential() >= differentialThreshold ||
                        stockPerformance.volumeDifferential() >= differentialThreshold) {
                    stockPerformanceList.add(keyValue);
                }
            }
        }
        return stockPerformanceList.isEmpty() ? null : KeyValue.pair(null, stockPerformanceList);
    }

    @Override
    public void close() {
        //no-op
    }
}
